package com.learning.hadoop.sqoop.thread;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * ClassName: Master
 * Description:主线程
 * Date: 2018/4/18 10:43 【需求编号】
 *
 * @author Sam Sho
 * @version V1.0.0
 */
@Slf4j
public class Master {
    private volatile boolean flag = true;
    private final int size = 1000;
    private ArrayBlockingQueue<Object> queue = new ArrayBlockingQueue<>(size);
    private ConcurrentLinkedQueue queue2 = new ConcurrentLinkedQueue();

    public static void main(String[] args) {
        Master master = new Master();
        master.run();
    }

    /**
     * 运行
     */
    public void run() {
        startWorkers();
        getData();
    }

    /**
     * 停止
     */
    public void stop() {
        setFlag(false);
    }


    private void getData() {
        while (flag) {
            // TODO 去数据库查询批量查询数据
            List<Object> list = Lists.newArrayList();
            queue.addAll(list);
        }
    }

    private void startWorkers() {
        // 线程池大小
        int poolSize = 2;
        // 批量保存定义阈值
        int batchSize = 200;
        /** 自定义线程名称 **/
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("calMoney-thread-%d").build();
        /** 创建 guava ListenableFuture ，使用线程池**/
//        ExecutorService executorService1 = Executors.newSingleThreadExecutor();
//        ExecutorService executorService2 = Executors.newFixedThreadPool(poolSize);
//        ExecutorService executorService2 = Executors.newCachedThreadPool();

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(poolSize, poolSize, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), threadFactory);
        ListeningExecutorService executorService = MoreExecutors.listeningDecorator(threadPoolExecutor);
        /** 创建工作线程**/
        Worker worker = new Worker(queue, batchSize);

        /** 启动线程 **/
        ListenableFuture<?> future = executorService.submit(worker);
        ListenableFuture<?> future2 = executorService.submit(new Worker(queue, batchSize));
        /** 捕捉线程异常 **/
        Futures.catchingAsync(future, RuntimeException.class, ex -> {
            log.error("记入数据库时出错", ex);
            ListenableFuture<?> f = executorService.submit(worker);
            return (ListenableFuture<Object>) f;
        });

        MoreExecutors.shutdownAndAwaitTermination(executorService, 1, TimeUnit.SECONDS);

    }

    private void setFlag(boolean flag) {
        this.flag = flag;
    }


}
